//package com.education.realtime.app.dws;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONAware;
//import com.alibaba.fastjson.JSONObject;
//import com.education.realtime.app.BaseAppV1;
//import com.education.realtime.bean.TrafficCountBean;
//import com.education.realtime.util.AtguiguUtil;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.common.state.ValueState;
//import org.apache.flink.api.common.state.ValueStateDescriptor;
//import org.apache.flink.cep.CEP;
//import org.apache.flink.cep.PatternSelectFunction;
//import org.apache.flink.cep.PatternStream;
//import org.apache.flink.cep.PatternTimeoutFunction;
//import org.apache.flink.cep.pattern.Pattern;
//import org.apache.flink.cep.pattern.conditions.SimpleCondition;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.*;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.util.Collector;
//import org.apache.flink.util.OutputTag;
//
//import java.time.Duration;
//import java.util.*;
//
//public class DwsTrafficCount extends BaseAppV1 {
//    public static void main(String[] args) {
//        new DwsTrafficCount().init(3001,"DwsTrafficCount","dwd_traffic_page");
//    }
//    @Override
//    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
//        //1.做出uv
//        DataStreamSink<String> UVStream = pageToUV(stream);
//        //2.做出uj
//        DataStreamSink<String> UJStream = pageTOUJ(stream);
//        //3.union
//        SingleOutputStreamOperator<TrafficCountBean> unionStream = unionPageUVandUJ(stream,UVStream,UJStream)
//        //4.开窗聚合
//        //5.写到ck
//
//    }
//
//    private DataStream<TrafficCountBean> unionPageUVandUJ(DataStreamSource<String> stream,
//                                                             DataStreamSink<String> uvStream,
//                                                             DataStreamSink<String> ujStream) {
//        Map<String, DataStreamSink<String>> topicAndStreamMap = new HashMap<>();
//        topicAndStreamMap.put("page",stream.print());
//        topicAndStreamMap.put("uv",uvStream);
//        topicAndStreamMap.put("uj",ujStream);
//
//
//        SingleOutputStreamOperator<TrafficCountBean> pvSvDuringTimeStream = topicAndStreamMap
//                .get(TOPIC_DWD_TRAFFIC_PAGE)
//                .map(new MapFunction<String,TrafficCountBean>() {
//                    @Override
//                    public TrafficPageViewBean map(String value) throws Exception {
//                        JSONObject obj = JSON.parseObject(value);
//                        JSONObject common = obj.getJSONObject("common");
//                        JSONObject page = obj.getJSONObject("page");
//
//
//
//
//                        return new TrafficCountBean("", "",
//
//                        );
//
//                    }
//                });
//
//        SingleOutputStreamOperator<TrafficCountBean> uvStream = topicAndStreamMap
//                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
//                .map(json -> {
//
//                    JSONObject obj = JSON.parseObject(json);
//                    JSONObject common = obj.getJSONObject("common");
//                    JSONObject page = obj.getJSONObject("page");
//
//
//
//
//                    return new TrafficCountBean("", "",
//
//                    );
//
//
//                });
//
//
//        SingleOutputStreamOperator<TrafficCountBean> ujStream = topicAndStreamMap
//                .get(TOPIC_DWD_TRAFFIC_UJ_DETAIL)
//                .map(json -> {
//                    JSONObject obj = JSON.parseObject(json);
//                    JSONObject common = obj.getJSONObject("common");
//                    JSONObject page = obj.getJSONObject("page");
//
//
//
//
//                    return new TrafficCountBean("", "",
//
//                    );
//
//                });
//
//        return pvSvDuringTimeStream.union(ujStream, uvStream);
//    }
//
//    }
//
//    private DataStreamSink<String> pageTOUJ(DataStreamSource<String> stream) {
//        KeyedStream<JSONObject, String> keyedStream = stream
//                .map(JSON::parseObject)
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                                .withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
//                )
//                .keyBy(obj -> obj.getJSONObject("common").getString("mid"));
//
//        // 1. 定义规则
//        Pattern<JSONObject, JSONObject> pattern = Pattern
//                .<JSONObject>begin("entry1")
//                .where(new SimpleCondition<JSONObject>() {
//                    @Override
//                    public boolean filter(JSONObject value) throws Exception {
//                        String lastPageId = value.getJSONObject("page").getString("last_page_id");
//                        return lastPageId == null || lastPageId.length() == 0;
//                    }
//                })
//                .next("entry2")
//                .where(new SimpleCondition<JSONObject>() {
//                    @Override
//                    public boolean filter(JSONObject value) throws Exception {
//                        String lastPageId = value.getJSONObject("page").getString("last_page_id");
//                        return lastPageId == null || lastPageId.length() == 0;
//                    }
//                })
//                .within(Time.seconds(5));
//
//        // 2. 把规则作用到流上
//        PatternStream<JSONObject> ps = CEP.pattern(keyedStream, pattern);
//        // 3. 取出匹配或者超时的数据
//        SingleOutputStreamOperator<JSONObject> normal = ps.select(
//                new OutputTag<JSONObject>("late") {},
//                new PatternTimeoutFunction<JSONObject, JSONObject>() {
//
//                    @Override
//                    public JSONObject timeout(Map<String, List<JSONObject>> map,
//                                              long timeoutTimestamp) throws Exception {
//
//                        return map.get("entry1").get(0);
//                    }
//                },
//                new PatternSelectFunction<JSONObject, JSONObject>() {
//                    @Override
//                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
//                        return map.get("entry1").get(0);
//                    }
//                }
//        );
//
//
//       return normal.getSideOutput(new OutputTag<JSONObject>("late") {}).union(normal)
//                .map(JSONAware::toJSONString)
//                .print();
//    }
//
//    private DataStreamSink<String> pageToUV(DataStreamSource<String> stream) {
//        return stream
//                .map(JSON::parseObject)
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                                .withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
//
//                )
//                .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
//
//                    private ValueState<String> firstWindowState;
//
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        firstWindowState = getRuntimeContext()
//                                .getState(new ValueStateDescriptor<String>("firstWindowState", String.class));
//                    }
//
//                    @Override
//                    public void process(String key,
//                                        Context ctx,
//                                        Iterable<JSONObject> elements,
//                                        Collector<JSONObject> out) throws Exception {
//                        String yesterday = firstWindowState.value();
//                        String today = AtguiguUtil.toDate(ctx.window().getStart());
//
//                        if (!today.equals(yesterday)) {
//                            // 当天的第一个窗口
//                            firstWindowState.update(today);
//
//                            // 把窗口内所有的元素放入到list, 找出时间戳最小
//                            List<JSONObject> list = AtguiguUtil.toList(elements);
//
//                            // Collections.min(list, (o1, o2) -> o1.getLong("ts").compareTo(o2.getLong("ts")))
//                            JSONObject min = Collections.min(list, Comparator.comparing(o -> o.getLong("ts")));
//                            out.collect(min);
//                        }
//                    }
//                })
//                .map(JSONAware::toJSONString)
//                .print();
//    }
//}
